package db.prob.operators.projectjoin;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import db.prob.io.LineParser;
import db.prob.io.PTuple;

public class MapperProjectJoin extends Mapper<LongWritable, Text, Text, Text> {
	
	public static final String TABLE_1 = "table_1";
	public static final String COLUMNS_1 = "columns_1"; // for projection
	public static final String TABLE_1_ATTR = "table_1_attr"; // for join
	
	public static final String TABLE_2 = "table_2";
	public static final String COLUMNS_2 = "columns_2";
	public static final String TABLE_2_ATTR = "table_2_attr";

	public static final String RESULT_TABLE = "result_table";
	public static final String PASS = "pass";
	
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		PTuple tuple = LineParser.createPTuple(value.toString(), "\\t");
		Configuration conf = context.getConfiguration();

		String table1 = conf.get(TABLE_1);
		String table2 = conf.get(TABLE_2);

		Text newKey = new Text();
		Text newVal = new Text();
		
		if (tuple.getTableName().equals(table1)) {
			handleTable(conf, tuple, COLUMNS_1, TABLE_1_ATTR, newKey, newVal);
		} else if (tuple.getTableName().equals(table2)) {
			handleTable(conf, tuple, COLUMNS_2, TABLE_2_ATTR, newKey, newVal);
		} else {
			newKey.set(PASS);
			newVal = value;			
		}

		context.write(newKey,newVal);		
	}

	/**
	 * @param tuple
	 * @param conf
	 * @param newKey
	 * @param newVal
	 */
	private void handleTable(Configuration conf, PTuple tuple, String projColumns, String joinAttr, Text newKey, Text newVal) {
		String[] projAttrIndex = conf.get(projColumns).split(",");
		String[] projAttr = new String[projAttrIndex.length];
		for (int i = 0; i < projAttrIndex.length; i++) {
			// Creating the projected tuple's attributes.
			projAttr[i] = tuple.getAttr()[Integer.parseInt(projAttrIndex[i])];
		}
		PTuple newTuple = new PTuple(tuple.getTableName(), tuple.getProbability(), projAttr);
		int joinAttrInt = conf.getInt(joinAttr, Integer.MAX_VALUE);
		newKey.set(tuple.getAttr()[joinAttrInt]);
		newVal.set(newTuple.toString());
	}
	
	
	 
}
